package cn.itcast.action;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;
import java.util.UUID;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.action
 * @date 2021/3/8 0008 15:20
 * @Copyright © 2015-04-29  One for each, and two for each
 */
public class OrderAutomaticFavorableComments {
    public static void main(String[] args) {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);


        env.addSource(new MySource());

    }

    public static class MySource implements SourceFunction<Tuple3<String,String,Long>>{
        private boolean flag=true;
        @Override
        public void run(SourceContext<Tuple3<String, String, Long>> ctx) throws Exception {
            Random r =new Random();
            while(flag){
               String userId= r.nextInt(5)+ "";
               String orderId = UUID.randomUUID().toString();
               long currentTimeMillis = System.currentTimeMillis();
                ctx.collect(Tuple3.of(userId,orderId,currentTimeMillis));
               Thread.sleep(500);
            }
        }
        @Override
        public void cancel() {
            flag=false;
        }
    }
}
